package cc.eddic.practice.concurrency.impl;

import cc.eddic.practice.concurrency.LogItem;
import cc.eddic.practice.concurrency.LogParser;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import lombok.val;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

import static cc.eddic.practice.concurrency.ExecutorServiceUtil.shutdown;

@Slf4j
public class LogParserWithFuturesImpl implements LogParser {
    LogItemParserLhImpl logItemParserLh = new LogItemParserLhImpl();

    @Override
    public Map<String, List<LogItem>> parse(String folderName) {
        Map<String, List<LogItem>> map = new ConcurrentHashMap<>();
        ExecutorService executorService = Executors.newFixedThreadPool(16);
        List<Future<Model>> futures;
        try{
            val p=Paths.get(folderName);
            try(val walk=Files.walk(p)){
                futures=walk.parallel().filter(Files::isRegularFile)
                        .filter(path -> path.getFileName().toString().endsWith(".log"))
                        .map(model->executorService.submit(()->parse(model))).collect(Collectors.toList());
            }
        } catch (IOException e){
            log.error(e.getMessage());
            return map;
        }
        allOf(futures);
        futures.stream().parallel().map(this::getFuture).filter(Objects::nonNull)
                .forEach(x->map.put(x.getPath().toString(),x.getItems()));
        shutdown(executorService);
        return map;
    }
    private static <T> void waitDone(final List<Future<T>> futures){
        int cnt=0;
        while (cnt<futures.size()){
           for (val f:futures){
               if (f.isDone()) cnt++;
               if (log.isDebugEnabled()) log.debug("{} of {} done", cnt, futures.size());
           }
        }
    }

    private static <T> void allOf(final List<Future<T>> futures) {
        while (true) {
            val b = futures.stream().anyMatch(x -> !x.isDone());
            if (!b) break;

            sleep(10);
            if (log.isDebugEnabled()) log.debug("{} futures done", futures.stream().filter(Future::isDone).count());
        }
    }

    private static void sleep(int millis){
        try {
            Thread.sleep(millis);
        }catch (InterruptedException e){
            log.error(e.getMessage());
        }
    }
    protected Model parse(Path path){
        try (val lines = Files.lines(path)){
            val items = lines.map(logItemParserLh::parse)
                    .filter(Optional::isPresent)
                    .map(Optional::get)
                    .collect(Collectors.toList());
            if (log.isDebugEnabled()) log.debug("PARSE : {} => {}", path, items .size());
            return new Model(path, items);
        }catch (IOException e){
            log.error(e.getMessage(),e);
            return new Model(path);
        }
    }
    private Model getFuture(Future<Model> future){
        try {
            return future.get();
        } catch (ExecutionException |InterruptedException e) {
           log.error(e.getMessage());
           return null;
        }
    }
    @Getter
    protected static class  Model{
        public Model(Path path){
            this(path,new ArrayList<>());
        }

        public Model(Path path, List<LogItem> items) {
            this.path = path;
            this.items = items;
        }

        private  final  Path path;
        private final List<LogItem> items;
    }

}